package com.bingxu.flink.action;

import com.bingxu.flink.bean.AdsClickLog;
import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class Click_Statistics {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        // load data and transform Java Bean object
        DataStreamSource<String> sourceDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\AdClickLog.csv");
        SingleOutputStreamOperator<AdsClickLog> adsClickDS = sourceDS.map(new MapFunction<String, AdsClickLog>() {
            @Override
            public AdsClickLog map(String s) throws Exception {
                String[] fields = s.split(",");
                return new AdsClickLog(
                        Long.parseLong(fields[0]),
                        Long.parseLong(fields[1]),
                        fields[2],
                        fields[3],
                        Long.parseLong(fields[4])
                );
            }
        });
        // 数据处理
        adsClickDS
                .map(ele -> Tuple2.of(ele.getProvince() + ":" + ele.getAdId(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(item -> item.f0)
                .sum(1)
                .print();

        // 运行执行
        env.execute();

    }
}
